package blue.mqtt4j.internal.client.impl;

import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.NavigableSet;
import java.util.concurrent.BlockingQueue;

/**
 * @author Jin Zheng
 * @since 1.0 2020-10-23
 */
public class PublisherListener implements ChannelFutureListener
{
	private static Logger logger = LoggerFactory.getLogger(PublisherListener.class);

	private final PublishMessage message;
	private BlockingQueue<PublishMessage> queue;
	private NavigableSet<Integer> pubAckSet;
	private NavigableSet<Integer> pubRecSet;
	private NavigableSet<Integer> pubCompSet;

	private PublisherListener(PublishMessage message)
	{
		this.message = message;
	}

	public static PublisherListener createWithQoS0(PublishMessage message, BlockingQueue<PublishMessage> queue)
	{
		PublisherListener listener = new PublisherListener(message);
		listener.queue = queue;
		return listener;
	}

	public static PublisherListener createWithQoS1(PublishMessage message, NavigableSet<Integer> pubAckSet)
	{
		PublisherListener listener = new PublisherListener(message);
		listener.pubAckSet = pubAckSet;
		return listener;
	}

	public static PublisherListener createWithQoS2(PublishMessage message, NavigableSet<Integer> pubRecSet)
	{
		PublisherListener listener = new PublisherListener(message);
		listener.pubRecSet = pubRecSet;
		return listener;
	}

	public static PublisherListener createWithPubRel(PublishMessage message, NavigableSet<Integer> pubCompSet)
	{
		PublisherListener listener = new PublisherListener(message);
		listener.pubCompSet = pubCompSet;
		return listener;
	}

	@Override
	public void operationComplete(ChannelFuture future) throws Exception
	{
		if (future.isSuccess())
		{
			logger.debug("Send successful, type: {}, messageId: {}", message.type(), message.messageId());
			switch (message.type())
			{
				case PUBREL:
					if (pubCompSet != null)
					{
						pubCompSet.remove(message.messageId());
					}
					break;
				case PUBLISH:
					switch (message.qos())
					{
						case AT_LEAST_ONCE:
							if (pubAckSet != null)
							{
								pubAckSet.remove(message.messageId());
							}
							break;
						case EXACTLY_ONCE:
							if (pubRecSet != null)
							{
								pubRecSet.remove(message.messageId());
							}
							break;
						default:
							break;
					}
					break;
				default:
					break;
			}

		}
		else
		{
			logger.error("Send failure");
			if (queue != null)
			{
				switch (message.qos())
				{
					case AT_MOST_ONCE:
						queue.offer(message);
						break;
					default:
						break;
				}
			}
		}
	}
}
